package drds.binlog.parse.inbound.mysql.tsdb;

import com.alibaba.fastsql.sql.ast.SQLDataType;
import com.alibaba.fastsql.sql.ast.SQLDataTypeImpl;
import com.alibaba.fastsql.sql.ast.SQLExpr;
import com.alibaba.fastsql.sql.ast.SQLStatement;
import com.alibaba.fastsql.sql.ast.expr.*;
import com.alibaba.fastsql.sql.ast.statement.*;
import com.alibaba.fastsql.sql.dialect.mysql.ast.MySqlPrimaryKey;
import com.alibaba.fastsql.sql.dialect.mysql.ast.MySqlUnique;
import com.alibaba.fastsql.sql.repository.Schema;
import com.alibaba.fastsql.sql.repository.SchemaObject;
import com.alibaba.fastsql.sql.repository.SchemaRepository;
import com.alibaba.fastsql.util.JdbcConstants;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.parse.inbound.FieldMetaData;
import drds.binlog.parse.inbound.TableMetaData;
import drds.binlog.parse.inbound.mysql.ddl.DruidDdlParser;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于DDL维护的内存表结构
 */
public class MemoryTableMeta implements TableMetaTSDB
{

    private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class);
    private Map<List<String>, TableMetaData> tableMetas = new ConcurrentHashMap<List<String>, TableMetaData>();
    private SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL);

    public MemoryTableMeta()
    {
    }

    @Override
    public boolean init(String destination)
    {
        return true;
    }

    public boolean apply(EntryPosition position, String schema, String ddl, String extra)
    {
        tableMetas.clear();
        synchronized (this)
        {
            if (StringUtils.isNotEmpty(schema))
            {
                repository.setDefaultSchema(schema);
            }

            try
            {
                // druid暂时flush privileges语法解析有问题
                if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush")
                        && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "grant"))
                {
                    repository.console(ddl);
                }
            } catch (Throwable e)
            {
                logger.warn("parse faield : " + ddl, e);
            }
        }

        // TableMetaData metadata = find("tddl5_00", "ab");
        // if (metadata != null) {
        // repository.setDefaultSchema("tddl5_00");
        // System.out.println(repository.console("show create table tddl5_00.ab"));
        // System.out.println(repository.console("show columns from tddl5_00.ab"));
        // }
        return true;
    }

    @Override
    public TableMetaData find(String schema, String table)
    {
        List<String> keys = Arrays.asList(schema, table);
        TableMetaData tableMetaData = tableMetas.get(keys);
        if (tableMetaData == null)
        {
            synchronized (this)
            {
                tableMetaData = tableMetas.get(keys);
                if (tableMetaData == null)
                {
                    Schema schemaRep = repository.findSchema(schema);
                    if (schemaRep == null)
                    {
                        return null;
                    }
                    SchemaObject data = schemaRep.findTable(table);
                    if (data == null)
                    {
                        return null;
                    }
                    SQLStatement statement = data.getStatement();
                    if (statement == null)
                    {
                        return null;
                    }
                    if (statement instanceof SQLCreateTableStatement)
                    {
                        tableMetaData = parse((SQLCreateTableStatement) statement);
                    }
                    if (tableMetaData != null)
                    {
                        if (table != null)
                        {
                            tableMetaData.setTableName(table);
                        }
                        if (schema != null)
                        {
                            tableMetaData.setSchemaName(schema);
                        }

                        tableMetas.put(keys, tableMetaData);
                    }
                }
            }
        }

        return tableMetaData;
    }

    @Override
    public boolean rollback(EntryPosition position)
    {
        throw new RuntimeException("not support for memory");
    }

    public Map<String, String> snapshot()
    {
        Map<String, String> schemaDdls = new HashMap<String, String>();
        for (Schema schema : repository.getSchemas())
        {
            StringBuffer data = new StringBuffer(4 * 1024);
            for (String table : schema.showTables())
            {
                SchemaObject schemaObject = schema.findTable(table);
                schemaObject.getStatement().output(data);
                data.append("; \n");
            }
            schemaDdls.put(schema.getName(), data.toString());
        }

        return schemaDdls;
    }

    private TableMetaData parse(SQLCreateTableStatement statement)
    {
        int size = statement.getTableElementList().size();
        if (size > 0)
        {
            TableMetaData tableMetaData = new TableMetaData();
            for (int i = 0; i < size; ++i)
            {
                SQLTableElement element = statement.getTableElementList().get(i);
                processTableElement(element, tableMetaData);
            }
            return tableMetaData;
        }

        return null;
    }

    private void processTableElement(SQLTableElement element, TableMetaData tableMetaData)
    {
        if (element instanceof SQLColumnDefinition)
        {
            FieldMetaData fieldMetaData = new FieldMetaData();
            SQLColumnDefinition column = (SQLColumnDefinition) element;
            String name = getSqlName(column.getName());
            // String charset = getSqlName(column.getCharsetExpr());
            SQLDataType dataType = column.getDataType();
            String dataTypStr = dataType.getName();
            if (dataType.getArguments().size() > 0)
            {
                dataTypStr += "(";
                for (int i = 0; i < column.getDataType().getArguments().size(); i++)
                {
                    if (i != 0)
                    {
                        dataTypStr += ",";
                    }
                    SQLExpr arg = column.getDataType().getArguments().get(i);
                    dataTypStr += arg.toString();
                }
                dataTypStr += ")";
            }

            if (dataType instanceof SQLDataTypeImpl)
            {
                SQLDataTypeImpl dataTypeImpl = (SQLDataTypeImpl) dataType;
                if (dataTypeImpl.isUnsigned())
                {
                    dataTypStr += " unsigned";
                }

                if (dataTypeImpl.isZerofill())
                {
                    dataTypStr += " zerofill";
                }
            }

            if (column.getDefaultExpr() == null || column.getDefaultExpr() instanceof SQLNullExpr)
            {
                fieldMetaData.setDefaultValue(null);
            } else
            {
                fieldMetaData.setDefaultValue(DruidDdlParser.unescapeQuotaName(getSqlName(column.getDefaultExpr())));
            }

            fieldMetaData.setColumnName(name);
            fieldMetaData.setColumnType(dataTypStr);
            fieldMetaData.setNullable(true);
            List<SQLColumnConstraint> constraints = column.getConstraints();
            for (SQLColumnConstraint constraint : constraints)
            {
                if (constraint instanceof SQLNotNullConstraint)
                {
                    fieldMetaData.setNullable(false);
                } else if (constraint instanceof SQLNullConstraint)
                {
                    fieldMetaData.setNullable(true);
                } else if (constraint instanceof SQLColumnPrimaryKey)
                {
                    fieldMetaData.setPrimaryKey(true);
                } else if (constraint instanceof SQLColumnUniqueKey)
                {
                    fieldMetaData.setUniqueKey(true);
                }
            }
            tableMetaData.addFieldMeta(fieldMetaData);
        } else if (element instanceof MySqlPrimaryKey)
        {
            MySqlPrimaryKey column = (MySqlPrimaryKey) element;
            List<SQLSelectOrderByItem> pks = column.getColumns();
            for (SQLSelectOrderByItem pk : pks)
            {
                String name = getSqlName(pk.getExpr());
                FieldMetaData field = tableMetaData.getFieldMetaData(name);
                field.setPrimaryKey(true);
            }
        } else if (element instanceof MySqlUnique)
        {
            MySqlUnique column = (MySqlUnique) element;
            List<SQLSelectOrderByItem> uks = column.getColumns();
            for (SQLSelectOrderByItem uk : uks)
            {
                String name = getSqlName(uk.getExpr());
                FieldMetaData field = tableMetaData.getFieldMetaData(name);
                field.setUniqueKey(true);
            }
        }
    }

    private String getSqlName(SQLExpr sqlName)
    {
        if (sqlName == null)
        {
            return null;
        }

        if (sqlName instanceof SQLPropertyExpr)
        {
            SQLIdentifierExpr owner = (SQLIdentifierExpr) ((SQLPropertyExpr) sqlName).getOwner();
            return DruidDdlParser.unescapeName(owner.getName()) + "."
                    + DruidDdlParser.unescapeName(((SQLPropertyExpr) sqlName).getName());
        } else if (sqlName instanceof SQLIdentifierExpr)
        {
            return DruidDdlParser.unescapeName(((SQLIdentifierExpr) sqlName).getName());
        } else if (sqlName instanceof SQLCharExpr)
        {
            return ((SQLCharExpr) sqlName).getText();
        } else if (sqlName instanceof SQLMethodInvokeExpr)
        {
            return DruidDdlParser.unescapeName(((SQLMethodInvokeExpr) sqlName).getMethodName());
        } else
        {
            return sqlName.toString();
        }
    }

    public SchemaRepository getRepository()
    {
        return repository;
    }

}
